package com.dtruth.dataflow.bolt;

import java.util.Map;

import org.slf4j.Logger;

import com.dtruth.dataflow.dao.RedisDAO;
import com.dtruth.dataflow.util.Env;
import com.dtruth.dataflow.util.MyLoggerFactory;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class DataRefererBolt extends BaseBasicBolt {

	private static Logger log = MyLoggerFactory
			.getLogger(DataRefererBolt.class);
	private static final long serialVersionUID = 1L;

	RedisDAO rd = null;

	/**
	 * On create
	 */
	@Override
	public void prepare(Map stormConf, TopologyContext context) {
		String host = Env.getInstance().getProperty("redis.dataconfig.host")
				.trim();
		int port = Integer.valueOf(Env.getInstance()
				.getProperty("redis.dataconfig.port").trim());
		int insNum = Integer.valueOf(Env.getInstance()
				.getProperty("redis.dataconfig.db").trim());
		rd = new RedisDAO(host, port, insNum);
	}

	public void execute(Tuple input, BasicOutputCollector collector) {
		@SuppressWarnings("unchecked")
		Map<String, Object> map = (Map<String, Object>) input.getValue(0);
		Object obj = map.get("referer");
		if (obj != null) {
			String referer = obj.toString();
			Map<String, String> rinfo = rd.hgetAll("list:" + referer);
			if (rinfo != null) {
				String channel = rinfo.get("channel");
				if (channel != null)
					map.put("channel", channel);
				String quality = rinfo.get("quality");
				if (quality != null)
					map.put("quality", quality);
				else
					map.put("quality", 5);
			}
		}
		log.debug("emit:" + map.get("url"));
		collector.emit(new Values(map));
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("news"));
	}
}